package com.github.codemperor.core.logger.compute.mapreduce;

import com.github.codemperor.core.logger.compute.Context;
import com.github.codemperor.core.logger.compute.IMapperStreamingFunction;
import com.github.codemperor.core.logger.compute.IReducerStreamingFunction;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * 日志文件处理解析的抽象类
 * 利用MapReduce的机制
 */
public class MapReduceProcess {
    private MapReduceProcess() {
    }

    /**
     * mapper实现类
     */
    private IMapperStreamingFunction mapperClass;

    public MapReduceProcess setMapperClass(IMapperStreamingFunction mapperClass) {
        this.mapperClass = mapperClass;
        return this;
    }

    /**
     * 计算实现类
     */
    private IReducerStreamingFunction reducerClass;

    public MapReduceProcess setReducerClass(IReducerStreamingFunction reducerClass) {
        this.reducerClass = reducerClass;
        return this;
    }

    public static MapReduceProcess getInstance() {
        return new MapReduceProcess();
    }


    /**
     * 数据进行切分，切分成多个数据块（当然是分布式分散在不同节点啦）
     * 不过这个就不做分布式了，只是将数据通过kv方式存储起来（相同key也认为是两个数据）
     */
    private void splitting() {
        //TODO 这个是为分布式的,暂时忽略
    }

    /**
     * 通过mapper，将数据相同的key整合在一个数据块上,并将同样的数据块计算命令发给分布式的reduce
     */
    private void shuffling(Object data, Context context) {
        mapperClass.mapping(data, context);
    }

    /**
     * 将所有相同的key的数据进行整合计算，最终得到结果
     */
    private void reducing(Object k, List<Object> values, Context context) {
        reducerClass.reducing(k, values, context);
    }

    /**
     * 执行方法
     */
    public Map<Object, Object> execute(List<String> dataList) {
        Context context = new Context();
        //List<String> dataList = new ArrayList<>();
        splitting();
        dataList.forEach(d -> shuffling(d, context));
        context.getMapperBlock().forEach((k, v) -> reducing(k, v, context));
        return context.getResult();
    }

    /**
     * 读取文件执行计算
     */
    public Map<Object, Object> execute(String path) throws IOException {
        BufferedReader in = new BufferedReader(new FileReader(path));
        List<String> dataList = new ArrayList<>();
        StringBuffer sb;
        while (in.ready()) {
            sb = (new StringBuffer(in.readLine()));
            if (!"".equals(sb.toString().trim())) {
                dataList.add(sb.toString());
            }
        }
        in.close();
        return execute(dataList);
    }
}
